package edu.hawaii.ics.pjdb.partition_function;

import java.util.ArrayList;
import java.util.List;

import edu.hawaii.ics.pjdb.types.Range;
import edu.hawaii.ics.pjdb.types.Tuple;

/**
 * Defines a PartitionFunction which partitions the given tuples based on the
 * provided ranges of values.
 * 
 * @author Christopher Foo
 * 
 */
public class RangePartition implements PartitionFunction {

    /**
     * Gets the ranges from the List of argument Strings. Will discard the last
     * value if given an odd number of argument Strings.
     * 
     * @param arguments
     *            The argument Strings representing the ranges. Every even index
     *            (including 0) will be the start of the range and every odd
     *            index will be the end of the range.
     * @return The ranges generated from the arguments.
     */
    public List<Range> getRanges(List<String> arguments) {
        ArrayList<Range> ranges = new ArrayList<Range>();
        int startIndex = 0;
        int endIndex = 1;
        int numArguments = arguments.size();
        for (; endIndex < numArguments; startIndex = startIndex + 2, endIndex = endIndex + 2) {
            ranges.add(new Range().setStartValue(arguments.get(startIndex))
                    .setEndValue(arguments.get(endIndex)));
        }
        return ranges;
    }

    /**
     * Returns a fixed list of ranges where all starting values are less than
     * the ending values.
     * 
     * @param ranges
     *            The list of ranges to be fixed.
     * @return The fixed list of ranges.
     */
    public List<Range> fixRanges(List<Range> ranges) {
        String start, end;
        List<Range> fixedRanges = new ArrayList<Range>(ranges);
        for (int i = 0; i < ranges.size(); i++) {
            start = ranges.get(i).getStartValue();
            end = ranges.get(i).getEndValue();

            // If start is larger than the end of the range, flip the start and
            // end.
            if (this.compareValues(start, end) > 0) {
                fixedRanges.get(i).setStartValue(end);
                fixedRanges.get(i).setEndValue(start);
            }
        }
        return fixedRanges;
    }

    /**
     * Compares the two Strings, returning a positive number if start > end, a
     * negative number if start < end, and 0 if start == end.
     * 
     * @param start
     *            The first String to be compared.
     * @param end
     *            The second String to be compared.
     * @return A positive number if start > end, a negative number if start <
     *         end, and 0 if start == end.
     */
    public int compareValues(String start, String end) {

        // See if String is actually an int.
        try {
            int startInt = Integer.parseInt(start);
            int endInt = Integer.parseInt(end);
            if (startInt < endInt) {
                return -1;
            }
            else if (startInt == endInt) {
                return 0;
            }
            else {
                return 1;
            }
        }

        // String is not an int.
        catch (NumberFormatException e) {
            // See if String is actually a double
            try {
                double startDouble = Double.parseDouble(start);
                double endDouble = Double.parseDouble(end);
                return Double.compare(startDouble, endDouble);
            }
            // String is not a double.
            catch (NumberFormatException e1) {
                // Handle as String
                return start.compareTo(end);
            }
        }
    }

    @Override
    /**
     * @param arguments List of Strings denoting the starting and ending values of the ranges to be used.  Even indexes denote starting values while odd indexes denote ending values.
     */
    public ArrayList<ArrayList<Tuple>> partition(List<Tuple> tuples,
            List<String> arguments, Integer partitionFieldIndex) {
        ArrayList<ArrayList<Tuple>> partitions = new ArrayList<ArrayList<Tuple>>();
        List<Range> ranges = this.fixRanges(this.getRanges(arguments));
        int numPartitions = ranges.size();

        // Initialize partitions with empty lists
        for (int i = 0; i < numPartitions; i++) {
            partitions.add(new ArrayList<Tuple>());
        }

        // Partition the tuples based on the ranges
        for (Tuple tuple : tuples) {
            int partitionBufferIndex = 0;
            for (Range range : ranges) {
                String start = range.getStartValue();
                String end = range.getEndValue();

                String partitionFieldValue = tuple.getEntries().get(
                        partitionFieldIndex);

                // Found range
                if (this.compareValues(partitionFieldValue, start) >= 0
                        && this.compareValues(partitionFieldValue, end) <= 0) {
                    partitions.get(partitionBufferIndex).add(tuple);
                    break;
                }
                partitionBufferIndex++;
            }
        }
        return partitions;
    }

}
